package com.mfs.proxy.core.server.reactor;

import com.mfs.common.constant.Constant;
import com.mfs.proxy.core.exceptions.ProxyException;
import com.mfs.proxy.core.server.config.ServerConfig;
import com.mfs.proxy.core.server.endpoint.DefaultProxyEndpoint;
import com.mfs.proxy.core.server.endpoint.ProxyEndpoint;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;

/**
 * 服务端代理任务
 * 用于监听代理端口，接收用户连接请求。
 * 收到连接请求后将建立的连接通道委托给 {@link com.mfs.proxy.core.server.reactor.SubProxyReactor}
 *
 * @author mfs
 * @date 2022/07/10
 * */
public class ProxyReactor implements Runnable {
    private final static Logger logger = LoggerFactory.getLogger(ProxyReactor.class);
    private static Selector proxySelector;
    public ProxyReactor() {
        try {
            proxySelector = Selector.open();
        } catch (IOException e) {
            throw new ProxyException(e);
        }
    }
    /**
     * 注册分配一个新的代理端点
     * @return ProxyEndpoint
     * */
    public static ProxyEndpoint register() {
        try {
            ServerSocketChannel proxyChannel = ServerSocketChannel.open();
            // 随机端口分配
            proxyChannel.bind(new InetSocketAddress(ServerConfig.getInstance().getProxyIp(), Constant.ZERO));
            proxyChannel.configureBlocking(false);
            SelectionKey selectionKey = proxyChannel.register(proxySelector.wakeup(), SelectionKey.OP_ACCEPT);
            return new DefaultProxyEndpoint(ServerConfig.getInstance().getProxyIp(), proxyChannel.socket().getLocalPort(), selectionKey);
        } catch (IOException e) {
            logger.debug("启动代理端点失败", e);
        }
        return null;
    }

    @Override
    public void run() {
        while (!Thread.currentThread().isInterrupted()) {
            try {
                int select = proxySelector.select(100);
                if (select < 1) {
                    Thread.yield();
                    continue;
                }
                Set<SelectionKey> selectionKeys = proxySelector.selectedKeys();
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
                while (iterator.hasNext()) {
                    SelectionKey selectionKey = iterator.next();
                    processAcceptable(selectionKey);
                    iterator.remove();
                }
            } catch (IOException e) {
                logger.error("处理连接请求失败", e);
            }
        }
    }

    /**
     * 处理连接请求，非连接请求会被忽略
     * */
    private void processAcceptable(SelectionKey selectionKey) {
        if (!(selectionKey.isValid() && selectionKey.isAcceptable())) {
            return;
        }
        ServerSocketChannel serverSocketChannel = (ServerSocketChannel) selectionKey.channel();
        SocketChannel userChannel = null;
        try {
            userChannel = serverSocketChannel.accept();
            userChannel.configureBlocking(false);
            SubProxyReactor.register(userChannel);
        } catch (IOException e) {
            logger.error("处理连接请求失败", e);
            if (userChannel != null) {
                try {
                    userChannel.close();
                } catch (IOException ioException) {
                    logger.error("关闭用户通道失败", e);
                }
            }
        }
    }
}
